#include "project2_include.h"

#define MAX_SHMSIZE SRINGSIZE*10
#define MAX_THREAD 5
#define MAX_REQUEST 1

static int verbose = 0;
static int num_requests = MAX_REQUEST;
double **request_time;

pthread_mutex_t global_mutex;
pthread_mutex_t *thread_mutex;
pthread_cond_t *thread_cond;
pthread_cond_t client_get_response;
int *response_available;
int can_get_response = 1;
disk_event_response_t **response_array;

pthread_t masterthread;
disk_event_response_t global_response;

int server_pid;
int shm_id;
__disk_sring_t *ring;
__disk_front_ring_t frontring;
int min_sector;
int max_sector;

char currentdir[256];

void SIGhandler(int); /* signal handler */

static void *client_request(void *idx)
{
	//int num_requests = *numofrequests;
	struct timeval time1,time2;
	int i;
	disk_event_request_t request;
	disk_event_response_t response;
	request.domain_pid = ring->domain_pid;
	//request.domain_tid = pthread_self();
	int index = *(int *)idx;
	request.domain_tid = index;

	int shm_key = ftok(currentdir, pthread_self() + index);
	int thread_shm_id = shmget(shm_key, 1024, 0644 | IPC_CREAT);
	if(thread_shm_id == -1)
	{
		fprintf(stderr, "The shmget call failed, error number %d, error%s\n", errno, strerror(errno));
		exit(1);
	}
	unsigned long int *sector_ptr = shmat(thread_shm_id, NULL, 0); /* Attach segment by os in read and write mode */

	for(i=0; i<num_requests; i++)
	{
		//request.sector_number = request.domain_tid;
		request.sector_number = rand() % (max_sector - min_sector) + min_sector;
		request.shm_id = thread_shm_id;//printf("thread_shm_id:%d\n", thread_shm_id);
		gettimeofday(&time1, NULL);
		disk_event_put_request(&frontring, &request);
		//disk_event_get_response(&frontring, &response);
		pthread_mutex_lock(&thread_mutex[index]);
		//pthread_mutex_lock(&global_mutex);
		while(response_available[index] == 0){
			pthread_cond_wait(&thread_cond[index], &thread_mutex[index]);
			//pthread_cond_wait(&thread_cond[index], &global_mutex);
		}
		pthread_mutex_unlock(&thread_mutex[index]);
		gettimeofday(&time2, NULL);
		//can_get_response = 0;
		//printf("tid:%d;sector_number:%d\n", (int)pthread_self(), response.sector_number);
		//printf("tid:%d;sector_number:%d\n", response.domain_tid, response.sector_number);
		//printf("tid:%d;sector_number:%d\n", global_response.domain_tid, global_response.sector_number);
		//printf("tid:%d;sector_number:%d\n", response_array[index]->domain_tid, response_array[index]->sector_number);
		response_available[index] = 0;
		//can_get_response = 1;
		//pthread_mutex_unlock(&global_mutex);
		//pthread_cond_signal(&client_get_response);
		request_time[index][i] = (time2.tv_sec - time1.tv_sec) + (time2.tv_usec - time1.tv_usec)/1000000.0;
		//outputToFile(request.sector_number, response_array[index]->sector);
		outputToFile(request.sector_number, sector_ptr);
		//printf("shm:%d\n", sector_ptr[0]);
		//printf("shm:%d, ring:%d\n", sector_ptr[0], response_array[index]->sector[0]);
	}
	shmctl(thread_shm_id, IPC_RMID, 0);
}

void* signal_handler(int a, siginfo_t * info, void * b) {
	printf("received signal from %d \n", info->si_pid);
	return NULL;
}

int install_signal_handler() {
	struct sigaction act;
	act.sa_handler = signal_handler;
	act.sa_flags = (SA_ONSTACK | SA_RESTART | SA_SIGINFO);
	if (sigaction(SIGUSR2, &act, NULL)) {
		printf("FATAL: sigusr2 install failed !!");
		return -1;
	}

	printf("signal handler installed on process id: %d \n", getpid());

	return 0;
}

static void *signal_thread(void* argu) {
	disk_event_response_t *response;
	/*response = (disk_event_response_t *)malloc(sizeof(disk_event_response_t));
	int result;
	while(1)
	{
		result = disk_event_get_response(&frontring, response);
		if(result == -1)
		{
			//printf("No response\n");
		}
		else
		{
			response_array[response->domain_tid] = response;
			printf("domain_tid:%d,sector_number:%d\n", response->domain_tid, response->sector_number);
			response_available[response->domain_tid] = 1;
			pthread_cond_signal(&thread_cond[response->domain_tid]);
		}
	}*/
	while(1)
	{//printf("(_r)->sring->rsp_prod:%d, (_r)->rsp_cons;%d\n", frontring.sring->rsp_prod, frontring.rsp_cons);
		if(RING_HAS_UNCONSUMED_RESPONSES(&frontring) == 0) // no responses on the ring
		{
			//printf("No response\n");
		}
		else
		{
			//pthread_mutex_lock(&global_mutex);
			//while(can_get_response == 0) {
			//	pthread_cond_wait(&client_get_response, &global_mutex);
			//}
			response = (disk_event_response_t *)malloc(sizeof(disk_event_response_t));
			disk_event_get_response(&frontring, response);
			response_array[response->domain_tid] = response;
			//disk_event_get_response(&frontring, &global_response);
			//printf("domain_tid:%d,sector_number:%d\n", global_response.domain_tid, global_response.sector_number);
			////////printf("domain_tid:%d,sector_number:%d\n", response->domain_tid, response->sector_number);
			//memcpy(response_array[global_response.domain_tid], sizeof(global_response));
			response_available[response->domain_tid] = 1;
			//pthread_mutex_unlock(&global_mutex);
			//pthread_cond_signal(&thread_cond[global_response.domain_tid]);
			pthread_cond_signal(&thread_cond[response->domain_tid]);
		}
	}
	/*unblock_signal(SIGUSR2);
	install_signal_handler();
	int i;
	int next_thread = 0; 
	//int total = (int) argu;
	sigset_t waitset;
	int sig;
	sigemptyset(&waitset);
	sigaddset(&waitset, SIGUSR2);
	//printf("total iteration: %d \n", total);
	while(1) {
		//printf("waiting for signal %d %u %d\n", i, waitset, total);
		if (sigwait(&waitset, &sig) != 0) {
			printf("sigwait() returned error number %d\n", errno);
			perror("sigwait() function failed\n");
		}
		fprintf(stderr, "received signal: %d \n ", sig);
		disk_event_get_response(&frontring, &response);
		pthread_mutex_lock(&arg[next_thread].count_mutex);
		pthread_cond_signal(&arg[next_thread].count_threshold_cv);
		pthread_mutex_unlock(&arg[next_thread].count_mutex);
		printf("unblocked thread: %d \n ", next_thread);
		next_thread++;
	}*/
}

int main (int argc, char *argv[])     
{
	struct timeval time1,time2;
        if(getcwd(currentdir, 256)==NULL)
        {
                fprintf(stderr, "Couldn't get current dir\n");
                exit(1);
        }
	initFiles();
	srand(time(NULL));
	int sockfd;
	char buffer[1024];
	struct sockaddr_in server_addr;
        struct hostent *host = gethostbyname("127.0.0.1");
	int portnumber = 8080;
	int num_threads = MAX_THREAD;
	int nbytes;
        int i, j;
	/*for (i=1; i<argc; i++)
	{
		char* argi = argv[i];
		if (!strcmp(argi, "-port") || !strcmp(argi, "-p"))
		{
			portnumber = strtod(argv[++i], 0);
		}
		else if (!strcmp(argi, "-host") || !strcmp(argi, "-h"))
		{
			host = gethostbyname(argv[++i]);
		}
		else if (!strcmp(argi, "-thread") || !strcmp(argi, "-t"))
		{
			num_threads = strtod(argv[++i], 0);
		}
		else if (!strcmp(argi, "-request") || !strcmp(argi, "-r"))
		{
			num_requests = strtod(argv[++i], 0);
		}
		else if (!strcmp(argi, "-v") || !strcmp(argi, "-verbose"))
		{
			verbose = strtod(argv[++i], 0);
		}
		else
		{
			fprintf(stderr, "Unrecognized argument\n");
			exit(1);
		}
	}*/
	if (argc != 3) {
		printf("FATAL: A client requires two arguments: num_of_threads and num_of_requests");
		exit(-1);
	}
	num_threads = strtod(argv[1], 0);
	num_requests = strtod(argv[2], 0)/num_threads;
	if(num_requests <= 0 )
	{
		printf("num_requests must be not smaller than num_threads\n");
		exit(0);
	}
	printf("threads:%d, requests per thread:%d\n", num_threads, num_requests);

	thread_mutex = (pthread_mutex_t *)malloc(sizeof(pthread_mutex_t)*num_threads);
	thread_cond = (pthread_cond_t *)malloc(sizeof(pthread_cond_t)*num_threads);
	response_available = (int *)malloc(sizeof(int)*num_threads);
	request_time = (double **)malloc(sizeof(double*)*num_threads);
	response_array = (disk_event_response_t **)malloc(sizeof(disk_event_response_t*)*num_threads);
	for(i=0; i<num_threads; i++)
	{
		pthread_mutex_init(&thread_mutex[i], NULL);
		pthread_cond_init(&thread_cond[i], NULL);
		response_available[i] = 0;
		request_time[i] = (double *)malloc(sizeof(double)*num_requests);
	}
	pthread_mutex_init(&global_mutex, NULL);
	pthread_cond_init(&client_get_response, NULL);

	if(host == NULL)
	{
		fprintf(stderr, "Gethostname error\n");
		exit (1);
	}
	//create socket
	if   ((sockfd   =   socket   (AF_INET,   SOCK_STREAM,   0))   ==   -1)   
	{        
		fprintf   (stderr,   "Socket   Error:%s\a\n",   strerror   (errno));
		exit   (1);            
	}
        
	//fill out server's info
	bzero   (&server_addr,   sizeof(server_addr));        
	server_addr.sin_family   =   AF_INET;      
	server_addr.sin_port   =   htons(portnumber);
	//server_addr.sin_addr   =   *((struct   in_addr   *)   host->h_addr);
        server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	//request connection
	if(connect(sockfd, (struct sockaddr *)(&server_addr), sizeof(struct sockaddr)) == -1)
	{
		fprintf(stderr, "Connect Error:%s\a\n", strerror(errno));
		exit (1);
	}

	int shm_key = ftok(currentdir, getpid());
	shm_id = shmget(shm_key, MAX_SHMSIZE, 0644 | IPC_CREAT);
	if(shm_id == -1)
	{
		fprintf(stderr, "The shmget call failed, error number %d, error%s\n", errno, strerror(errno));
		exit(1);
	}
	
	ring = shmat(shm_id, NULL, 0); /* Attach segment by os in read and write mode */
	if( ring == (void *) -1 || ring == NULL ){
		fprintf(stderr, "Fails to attach Shared Memory\n");
		exit (1);
	}
	SHARED_RING_INIT(ring);
	disk_event_ring_lock_init(ring);
	FRONT_RING_INIT(&frontring, ring, SRINGSIZE);

	ring->domain_pid = getpid();
	ring->shm_id = shm_id;

	char temp[256];
	sprintf(temp, "%d:%d", getpid(), shm_id);
	if(write(sockfd,  temp, strlen(temp)) == -1)
	{
		fprintf(stderr, "Write Error:%s\n", strerror(errno));
      		exit (1);
	}

	if((nbytes = read (sockfd, buffer, 1024))   ==   -1)
	{
		fprintf(stderr, "Read Error:%s\n", strerror(errno));
		exit (1);            
	}
	buffer[nbytes] = '\0';

	char * separator = strchr(buffer, ':');
	//unsigned int msg_id = strtod(separator+1, 0);
	*(separator) = '\0';
	server_pid = strtod(buffer, 0);
	char *buffer1 = separator+1;
	separator = strchr(buffer1, ':');
	*(separator) = '\0';
	unsigned int msg_id = strtod(buffer1, 0);
	buffer1 = separator+1;
	separator = strchr(buffer1, ':');
	*(separator) = '\0';
	min_sector = strtod(buffer1, 0);
	max_sector = strtod(separator+1, 0);

	printf("Server Pid:%d, msg_id:%d, min_sector:%d, max_sector:%d\n", server_pid, msg_id, min_sector, max_sector);
	close(sockfd);

	printf("msg_id:%d\n", ring->msg_id);
	ring->msg_id = msg_id;

	/* create thread handling signals */
	pthread_create(&masterthread, NULL, signal_thread, NULL);

	int error;
	int *index = (int *)malloc(sizeof(int)*num_threads);
	for(i=0; i<num_threads; i++)
		index[i] = i;

	pthread_t tid[num_threads];
	gettimeofday(&time1, NULL);
	for(i=0; i<num_threads; i++)
	{
		error = pthread_create(&tid[i], NULL, client_request, (void *)&index[0]); /* cannot use i directly since the address changes */
		index++;
		if(verbose)
		{
			if(0 != error)
				fprintf(stderr, "Couldn't run thread number %d, errno %d\n", i, error);
			else
				fprintf(stderr, "Thread tid%d\n", (int)tid[i]);
		}
	}

	/* now wait for all threads to terminate */
	for(i=0; i< num_threads; i++)
	{
		//void *status = 0;
		error = pthread_join(tid[i], NULL);
		if(verbose)
			fprintf(stderr, "Thread tid%d terminated\n", (int)tid[i]);
	}
	gettimeofday(&time2, NULL);

	//Tell server that client has done
	if   ((sockfd   =   socket   (AF_INET,   SOCK_STREAM,   0))   ==   -1)   
	{        
		fprintf   (stderr,   "Socket   Error:%s\a\n",   strerror   (errno));
		exit   (1);            
	}
	if(connect(sockfd, (struct sockaddr *)(&server_addr), sizeof(struct sockaddr)) == -1)
	{
		fprintf(stderr, "Connect Error:%s\a\n", strerror(errno));
		exit (1);
	}
	char finish[16];
	sprintf(finish, ":%d", getpid());
	if(write(sockfd,  finish, strlen(finish)) == -1)
	{
		fprintf(stderr, "Write Error:%s\n", strerror(errno));
      		exit (1);
	}
	close(sockfd);

	disk_event_ring_destroy(ring);
	shmctl(shm_id, IPC_RMID, 0);

	double elapsed_time = (time2.tv_sec - time1.tv_sec) + (time2.tv_usec - time1.tv_usec)/1000000.0;
	double max_request_time = 0;
	double min_request_time = request_time[0][0];
	double mean_request_time = 0;
	double dev_request_time = 0;
	for(i=0; i<num_threads; i++)
	{
		for(j=0; j<num_requests; j++)
		{
			if(max_request_time < request_time[i][j])
				max_request_time = request_time[i][j];
			if(min_request_time > request_time[i][j])
				min_request_time = request_time[i][j];
			mean_request_time = mean_request_time + request_time[i][j];
		}
	}
	mean_request_time = mean_request_time/(num_requests*num_threads);

	for(i=0; i<num_threads; i++)
	{
		for(j=0; j<num_requests; j++)
		{
			dev_request_time = dev_request_time + (request_time[i][j]-mean_request_time)*(request_time[i][j]-mean_request_time);
		}
	}
	dev_request_time = sqrt(dev_request_time/(num_requests*num_threads));

	max_request_time = max_request_time*1000000000;
	mean_request_time = mean_request_time*1000000000;
	min_request_time = min_request_time*1000000000;
	dev_request_time = dev_request_time*1000000000;
	double request_per_sec = (num_requests*num_threads)/elapsed_time;
	printf("%ld|%ld|%ld|%ld|%f\n", (long)max_request_time, (long)mean_request_time, (long)min_request_time, (long)dev_request_time, request_per_sec);
	closeFiles();
	exit (0);
}

void  SIGhandler(int sig)
{
	int i;
	char c;
	signal(sig, SIG_IGN);
	if(sig == SIGSEGV){
		fprintf(stderr, "user defined: Segmentataion Fault\n");
	}
	if(sig == SIGINT){
		fprintf(stderr, "Ctrl + c termination\n");
	}
	if( ring != (void *) -1 && ring != NULL )
	{
		disk_event_ring_destroy(ring);
		shmctl(ring->shm_id, IPC_RMID, 0);
	}
	closeFiles();
	exit(0);
}
